package edu.hit.stu;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;

import java.util.concurrent.TimeUnit;

public class IFlux {

    public static void main(String[] args) {
        Flux<Integer> integerFlux = Flux.just(1,2,3,4,5,6,7,8);

        ISubscribe iSubscribe = new ISubscribe();
        integerFlux.subscribe(iSubscribe);

        Flux<Integer> filter = integerFlux.filter(x -> x < 4);
        filter.subscribe(iSubscribe);
        integerFlux.subscribe(iSubscribe);
        filter.subscribe(iSubscribe);

        try {
            TimeUnit.SECONDS.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class ISubscribe implements Subscriber<Integer> {
    private int count = 0;
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        String msg = "开始订阅！";
        printWithColor(msg);
        this.subscription.request(1);
    }

    @Override
    public void onNext(Integer item) {
        count++;
        System.out.println("第" + count + "次接收到的值为: " + item);
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        this.subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        String msg = "出错啦！！";
        printWithColor(msg);
        throwable.printStackTrace();
    }

    @Override
    public void onComplete() {
        String msg = "订阅完成！！";
        printWithColor(msg);
    }

    private void printWithColor(String msg) {
        System.out.println("[107;38;5;88;1;4;6m" + msg + "[0m");
    }
}